package com.dwd;


import com.alibaba.fastjson.JSONObject;
import com.utils.KafkaUtils;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

/**
 * 在 DWD 层中，使用状态编程识别出新老用户。如果用户首次下单，则标记为新用户；否则标记为老用户，并在输出流中发送到相应的 Kafka 主题中
 */
public class NewOrOldUser {
    public static void main(String[] args) throws Exception {
        //流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //读取user_info
        FlinkKafkaConsumer<String> dwd_orderInfo_db = KafkaUtils.createConsumer("dwd_orderInfo_db","haha");
        DataStreamSource<String> orderInfoStream = env.addSource(dwd_orderInfo_db);
        //转为jsonObject

        SingleOutputStreamOperator<JSONObject> stream = orderInfoStream.map(x -> {
            return JSONObject.parseObject(x);
        });

        //识别新老用户
        KeyedStream<JSONObject, String> kunkun = stream.keyBy(x -> {
            return x.getJSONObject("after").getString("user_id");
        });

        kunkun.print();
        SingleOutputStreamOperator<JSONObject> newOrOldUser = kunkun
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> state;

            @Override
            public void open(Configuration parameters) throws Exception {
                //监控状态
                state = getRuntimeContext().getState(new ValueStateDescriptor<String>("state", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                //判断监控状态

                if (state.value()== null) {
                    state.update("新用户");
                    jsonObject.put("is_new", "1");
                } else {
                    jsonObject.put("is_new", "0");
                }
                collector.collect(jsonObject);
            }
        });
        newOrOldUser.print("kunkun>>>");

        newOrOldUser.map(x->JSONObject.toJSONString(x)).addSink(KafkaUtils.createProduer("dwd_newOrOld_db"));
        env.execute();

    }
}
